package com.atguigu.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class DwdTrafficUniqueVisitorDetail {
    public static void main(String[] args) throws Exception {
        // TODO: 2022/5/22 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // TODO: 2022/5/22 检查点相关设置
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(1000L * 60);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs:hadoop102:8020/xxxx");
        // TODO: 2022/5/22  从卡夫卡读取数据
        //声明消费者组和主题
        String topic = "dwd_traffic_page_log";
        String group = "dwd_traffic_page_log_group";
        //封装流
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, group));

        // TODO: 2022/5/22  对读取数据进行类型转换 jsonstr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

        jsonObjDS.print();

        // TODO: 2022/5/22  分组
        //KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("commom").getString("mid"));
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        // TODO: 2022/5/22  状态编程过滤出独立访客
        SingleOutputStreamOperator<JSONObject> uvDS = keyedDS.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<String> lastVisitDateState;
            //private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                //sdf = new SimpleDateFormat("yy-MM-dd");
                ValueStateDescriptor<String> lastVisitDateStateDesc =
                        new ValueStateDescriptor<>("lastVisitDateState", String.class);
                //设置状态的过期时间和过期的方式
                lastVisitDateStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());

                lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDesc);

            }

            @Override
            public boolean filter(JSONObject jsonObj) throws Exception {
                String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                //先过滤page_last_id不为空的数据
                if (lastPageId != null && lastPageId.length() > 0) {
                    return false;
                }
                //获取状态中的日期
                String lastVisitDate = lastVisitDateState.value();
                //获取数据中的当前日期
                String currVisitDate = DateFormatUtil.toDate(jsonObj.getLong("ts"));
                //判断状态中的值是否存在，如果存在且和当前数据相等 ，说明同一天重复，直接过滤
                if (lastVisitDate != null && lastVisitDate.length() > 0 && lastVisitDate.equals(currVisitDate)) {
                    return false;
                } else {
                    lastVisitDateState.update(currVisitDate);
                    return true;
                }


            }
        });

        uvDS.print("---------------");

        // TODO: 2022/5/22 再将独立访客保存再kafka主题中
        uvDS.map(jsonObj -> jsonObj.toJSONString())
            .addSink(MyKafkaUtil.getKafkaProducer("dwd_trffic_unique_visitor_detail"));


        env.execute();
    }
}
